package com.study.component;

import cn.hutool.core.util.RandomUtil;
import com.study.config.FastJsonCodec;
import com.study.dto.SubDTO;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

/**
 * @author zhuzi
 * 2024/3/26 18:22
 * 延迟队列
 */
@Slf4j(topic = "延迟队列")
//@Component
public class DelayedQueue {
    public static final String queueName = "redissonDelayedQueue";
    static RDelayedQueue<SubDTO> rDelayedQueue;
    @Resource
    RedissonClient redissonClient;

    //@PostConstruct
    public void init() {
        RBlockingQueue<SubDTO> blockingQueue = redissonClient.getBlockingQueue(queueName, new FastJsonCodec());
        rDelayedQueue = redissonClient.getDelayedQueue(blockingQueue);
    }


    /**
     * 每2秒执行一次
     */
    //@Scheduled(cron = "*/1 * * * * ?")
    public void pubData() {
        long delay = RandomUtil.randomLong(15L, 35L);
        SubDTO<Object> subDTO = new SubDTO<>().setType(RandomUtil.randomString(5)).setData("测试" + RandomUtil.randomString(5)).setPubDateTime(LocalDateTime.now());
        log.info("发布消息TTL={}|{}", delay, subDTO);
        rDelayedQueue.offer(subDTO, delay, TimeUnit.SECONDS);
    }


    //@Scheduled(cron = "*/1 * * * * ?")
    public void takeMsg() throws InterruptedException {
        RBlockingQueue<SubDTO> rBlockingQueue = redissonClient.getBlockingQueue(queueName, new FastJsonCodec());
        SubDTO msg = rBlockingQueue.take();
        log.info("读取消息     :{}", msg);
    }


}
